package cn.dansj.utils.project;

import cn.dansj.entity.User;
import cn.dansj.service.UserService;
import cn.dansj.common.context.ApplicationContextProvider;
import cn.dansj.common.utils.transfer.StringUtils;
import cn.dansj.common.utils.transfer.RandomUtils;
import cn.dansj.common.utils.transfer.Verification;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.stream.Collectors;

public class SSEUtils {
    private static final Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    private static final Logger logger = LoggerFactory.getLogger(SSEUtils.class);
    private static final UserService userService = ApplicationContextProvider.getBean(UserService.class);

    public static SseEmitter connect(String clientId) {
        if (clientId != null) {
            if (sseEmitterMap.containsKey(clientId)) {
                final SseEmitter sseEmitter = sseEmitterMap.get(clientId);
                sseEmitterMap.remove(clientId);
                sseEmitter.complete();
            }

            try {
                final SseEmitter sseEmitter = new SseEmitter(0L);
                sseEmitter.send(SseEmitter.event().id(RandomUtils.getUid()).data("连接成功"));
                sseEmitter.onCompletion(completionCallBack(clientId));
                sseEmitter.onError(errorCallBack(clientId));
                sseEmitter.onTimeout(timeoutCallBack(clientId));
                sseEmitterMap.put(clientId, sseEmitter);
                logger.info("创建SSE连接完成,当前用户:{},总用户数:{}", clientId, sseEmitterMap.size());
                return sseEmitter;
            } catch (Exception exception) {
                logger.info("创建SSE连接异常", exception);
            }
        }
        return null;
    }

    public static void sendMessage(String clientId, String message) {
        sendMessage(clientId, RandomUtils.getUid(), message, "", true);
    }

    public static void sendMessage(String clientId, String message, String eventName) {
        sendMessage(clientId, RandomUtils.getUid(), message, eventName, true);
    }

    public static void sendMessage(String clientId, String messageID, String message, String eventName, boolean like) {
        if (Verification.checkNull(clientId)) {
            for (String user : sseEmitterMap.keySet()) {
                if (Verification.checkNull(user)) continue;
                sendMessage(user, messageID, message, eventName, false);
            }
            return;
        }

        final Set<String> allClient = sseEmitterMap.keySet();
        List<String> sendClient = new ArrayList<>();
        for (String client : allClient) {
            if (like && (client.contains("@" + clientId) || client.contains(clientId + "@"))) {
                sendClient.add(client);
            } else if (!like && client.equals(clientId)) {
                sendClient.add(client);
            }
        }

        if (!sendClient.isEmpty()) {
            for (String client : sendClient) {
                SseEmitter sseEmitter = sseEmitterMap.get(client);
                try {
                    final SseEmitter.SseEventBuilder builder = SseEmitter.event().id(messageID).data(message);
                    if (Verification.checkNotNull(eventName)) builder.name(eventName);
                    sseEmitter.send(builder);
                    logger.info("用户{},消息ID:{},推送成功:{}", client, messageID, message);
                } catch (Exception exception) {
                    sseEmitterMap.remove(client);
                    sseEmitter.complete();
                    logger.info("用户：{}，消息ID：{}，推送异常：{}", client, messageID, exception.getMessage());
                }
            }
        } else {
            logger.info("未查找到类似客户端：{}", clientId);
        }
    }

    private static Runnable completionCallBack(String clientId) {
        return () -> {
            logger.info("结束SSE用户连接:{}", clientId);
            removeClient(clientId);
        };
    }

    private static Consumer<Throwable> errorCallBack(String clientId) {
        return throwable -> {
            logger.info("SSE用户连接异常：{}", clientId);
            removeClient(clientId);
        };
    }

    private static Runnable timeoutCallBack(String clientId) {
        return () -> {
            logger.info("连接SSE超时：{}", clientId);
            removeClient(clientId);
        };
    }

    public static void removeClient(String clientId) {
        if (clientId != null && sseEmitterMap.containsKey(clientId)) {
            final SseEmitter sseEmitter = sseEmitterMap.get(clientId);
            sseEmitterMap.remove(clientId);
            sseEmitter.complete();
        }
    }

    public static List<Map<String, String>> getUserName() {
        final List<User> users = userService.getUser();
        Map<String, String> userName = new HashMap<>();
        for (User user : users) {
            userName.put(user.getSequence(), user.getUsername());
        }
        return sseEmitterMap.keySet().stream().map(e -> split(e, "@", 2)).filter(Verification::checkNotNull).map(e -> {
            Map<String, String> userinfoMap = new HashMap<>();
            userinfoMap.put("username", userName.get(e));
            userinfoMap.put("sequence", e);
            return userinfoMap;
        }).distinct().collect(Collectors.toList());
    }

    public static String split(String val, String sep, int offset) {
        final String[] split = StringUtils.split(val, sep);
        return split.length >= offset ? split[offset - 1] : "";
    }
}
